package cn.ipanel.bigdata.dw.dim.hive

import cn.ipanel.bigdata.dw.dim.AbsDimHive
import cn.ipanel.bigdata.utils.Dictionary.{F_DEVICE_ID, F_REGION, F_STATUS}
import org.apache.spark.sql.{Dataset, Row}

/**
 * @author lzz
 * @environment IntelliJ IDEA 2020.3.1
 * @projectName bigdata_panyu
 * @date 2023/12/08 15:43
 * @description:  由于设备一次在线离线状态会跨天，需要按天记录全量设备前一天的在线状态，便于当天统计出结果
 */
protected[hive] object OnlineDeviceByDay extends AbsDimHive(tbName = "t_dim_device_online"){

  final val F_LAST_ONLINE_TIME            : String = "f_last_online_time"         // 记录目前为止最近一次在线时间，时间可能不是当天，格式：yyyy-MM-dd HH:mm:ss
  final val F_LAST_OFFLINE_TIME           : String = "f_last_offline_time"        // 记录目前为止最近一次离线时间，时间可能不是当天，格式：yyyy-MM-dd HH:mm:ss

  override def getTBColumns: Seq[String] = {
    Seq(F_DEVICE_ID, F_REGION, F_STATUS, F_LAST_ONLINE_TIME, F_LAST_OFFLINE_TIME, F_DATE)
  }

  override def buildTable: String = {
    s"""
       | CREATE TABLE IF NOT EXISTS $getDBName.$getTBName (
       |  $F_DEVICE_ID STRING,
       |  $F_REGION STRING,
       |  $F_STATUS TINYINT,
       |  $F_LAST_ONLINE_TIME STRING,
       |  $F_LAST_OFFLINE_TIME STRING
       | )
       | PARTITIONED by($F_DATE INT)
       | ROW FORMAT DELIMITED FIELDS TERMINATED BY $dataRowFormatter
       | MAP KEYS TERMINATED BY $dataMapFormatter
       | STORED AS $storageFormatter
       |""".stripMargin
  }

  override def emptyTable: Dataset[_] = {
    import spark.implicits._
    spark.createDataset(spark.sparkContext.emptyRDD[DeviceStatusTable])
  }

  case class DeviceStatusTable(var f_device_id: String
                               , var f_region: Long
                               , var f_status: Byte
                               , var f_last_online_time: String
                               , var f_last_offline_time: String
                               , var f_date: Int
                              )

  object DeviceStatusTable {
    def apply(row: Row): DeviceStatusTable = {
      DeviceStatusTable(
        row.getAs[String](F_DEVICE_ID)
        , row.getAs[Long](F_REGION)
        , row.getAs[Byte](F_STATUS)
        , row.getAs[String](F_LAST_ONLINE_TIME)
        , row.getAs[String](F_LAST_OFFLINE_TIME)
        , row.getAs[Int](F_DATE)
      )
    }
  }
}